package spring.cloud.tasks.executor.task.statistics;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.executor.common.Service;
import spring.cloud.tasks.executor.listener.config.TaskConfigurationService;
import spring.cloud.tasks.executor.task.TaskExecutor;
import spring.cloud.tasks.executor.utils.log.LogUtils;

import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 作业统计信息服务.
 */
@Slf4j
@Setter
@Getter
public class ProcessCountTaskService extends Service {
    private TaskConfigurationService taskConfigurationService;

    private ScheduledExecutorService scheduledExecutorService;

    private ScheduledFuture<?> scheduledFuture;

    private boolean isdown = false;

    public ProcessCountTaskService(final TaskExecutor taskExecutor) {
        super(taskExecutor);
    }

    @Override
    public synchronized void start() {
        taskConfigurationService = taskExecutor.getTaskConfigurationService();
        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
            private final AtomicInteger number = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable runnable) {
                StringBuilder name = new StringBuilder(taskConfiguration.getTaskId()).append("-ProcessCount-Thread-").append(number.incrementAndGet());
                Thread thread = new Thread(runnable, name.toString());
                thread.setDaemon(false);
                thread.setPriority(Thread.NORM_PRIORITY);
                return thread;
            }
        });
    }

    /**
     * 开启或重启统计处理数据数量的作业.
     */
    public synchronized void startProcessCountTask() {
        int processCountIntervalSeconds = taskConfigurationService.getProcessCountIntervalSeconds();
        if (processCountIntervalSeconds > 0) {
            if (scheduledFuture != null) {
                scheduledFuture.cancel(true);
                LogUtils.info(log, taskId, "Reschedule ProcessCountJob of the {} job, the processCountIntervalSeconds is {}", taskConfiguration.getTaskId(), processCountIntervalSeconds);
            }
            scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(new ProcessCountTask(taskExecutor), new Random().nextInt(10), processCountIntervalSeconds, TimeUnit.SECONDS);

        } else { // don't count, reset to zero.
            if (scheduledFuture != null) {
                LogUtils.info(log, taskId, "shutdown the task of reporting statistics data");
                scheduledFuture.cancel(true);
                scheduledFuture = null;
            }
        }
    }


    @Override
    public void stop() {
        if (isdown) {
            return;
        }
        isdown = true;
        stopProcessCountJob();
        ProcessCountStatistics.resetSuccessFailureCount(taskId, executorId);
    }

    /**
     * 停止统计处理数据数量的作业.
     */
    public synchronized void stopProcessCountJob() {
        if (scheduledFuture != null) {
            scheduledFuture.cancel(true);
        }
        if (scheduledExecutorService != null) {
            scheduledExecutorService.shutdown();
        }
    }
}
